package com.mingserve.raw.controller;

import com.alibaba.fastjson2.JSON;
import com.mingserve.common.core.constant.RawConstants;
import com.mingserve.common.core.domain.R;
import com.mingserve.common.core.utils.SpringUtils;
import com.mingserve.common.mqtt.listener.MessageListener;
import com.mingserve.common.mqtt.utils.MqttUtils;
import com.mingserve.raw.event.RawEvent;
import jakarta.servlet.http.HttpServletRequest;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.util.Map;

/**
 * 传感器数据上传接口API
 *
 * @author lvwshuai
 *
 * HTTP
 * Raw    原始包未编码：/raw/api/v1/upload
 * Stream 原始包Base64编码ZLib压缩: /raw/api/v1/stream/upload
 * Stream 原始包Base64编码ZLib压缩拆包: /raw/api/v1/stream/subpkg/upload
 * Raw    原始包Base64编码: /raw/api/v1/unzip/upload
 * Raw    原始包Base64编码: /raw/api/v1/unzip/subpkg/upload
 * RPM    原始转速包 /raw/api/v1/upload/rpm
 *
 * MQTT
 * Raw  原始包Base64编码ZLib压缩 raw/data/+
 * Raw  原始包Base64编码ZLib压缩拆包 raw/subpkg/+
 * Raw  原始包Base64编码 unzip/data/+
 * Raw  原始包Base64编码 unzip/subpkg/+
 * Conn 信息包 conn/info/+
 * RPM  原始转速包 raw/rpm/+
 *
 * 算法变转速
 * 算法转速 /raw/api/v1/upload/angular/rpm
 * 角域重采 /raw/api/v1/upload/angular
 */
@Slf4j
@RequiredArgsConstructor
@RestController
@RequestMapping("/raw/api/v1")
public class UploadController
{
    private final MqttUtils mqttUtils;

    /**
     * MQTT订阅传感器数据
     */
    public void subscribe()
    {
        subscribeToTopic(RawConstants.MQTT_RAW_ZIP + "/+", RawConstants.KAFKA_RAW_ZIP_TOPIC);
        subscribeToTopic(RawConstants.MQTT_RAW_SUB_ZIP + "/+", RawConstants.KAFKA_RAW_SUB_ZIP_TOPIC);
        subscribeToTopic(RawConstants.MQTT_RAW_CONN + "/+", RawConstants.KAFKA_RAW_DATA_TOPIC);
        subscribeToTopic(RawConstants.MQTT_RAW_UNZIP + "/+", RawConstants.KAFKA_RAW_UNZIP_TOPIC);
        subscribeToTopic(RawConstants.MQTT_RAW_SUB_UNZIP + "/+", RawConstants.KAFKA_RAW_SUB_UNZIP_TOPIC);
        subscribeToTopic(RawConstants.MQTT_RAW_RPM + "/+", RawConstants.KAFKA_RAW_PRM_TOPIC);
    }

    public void subscribeToTopic(String topicPattern, String kafkaTopic)
    {
        mqttUtils.subscribe(RawConstants.MQTT_DAQ, topicPattern, new MessageListener() {
            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                log.info("Received {} MQTT {} msg", RawConstants.MQTT_DAQ, topic);
                byte[] payload = mqttMessage.getPayload();
                Map<String, Object> parsedMap = null;
                byte[] rawBytes = null;
                try {
                    // 尝试解析 JSON，如果失败，则当作字节流处理
                    parsedMap = JSON.parseObject(payload);
                } catch (Exception e) {
                    rawBytes = payload;
                }
                publishRawEvent(kafkaTopic, topic, parsedMap, rawBytes);
            }
        });
    }


    /**
     * 接收传感器上传raw数据，写入 json 文件
     *
     * @param resultMap 原始数据
     * @return 处理结果
     */
    @PostMapping("/upload")
    public R<Void> upload(@RequestBody Map<String, Object> resultMap)
    {
        publishRawEvent(RawConstants.KAFKA_RAW_DATA_TOPIC, RawConstants.HTTP_UPLOAD, resultMap, null);
        return successResponse();
    }

    /**
     * 接收传感器上传的zlib压缩base64数据流-整包
     *
     * @param request
     * @return
     */
    @PostMapping("/stream/upload")
    public R<Void> uploadStream(HttpServletRequest request)
    {
        try {
            byte[] data = IOUtils.toByteArray(request.getInputStream());
            publishRawEvent(RawConstants.KAFKA_RAW_ZIP_TOPIC, RawConstants.HTTP_STREAM_ZIP, null, data);
            return successResponse();
        } catch (Exception ex) {
            log.error("Data stream conversion failed", ex);
            return R.fail("Data stream conversion failed");
        }
    }

    /**
     * 接收传感器上传的zlib压缩base64数据流-分包
     *
     * @param request
     * @return
     * @throws IOException
     */
    @PostMapping("/stream/subpkg/upload")
    public R<Void> uploadSubpackageStream(HttpServletRequest request)
    {
        try {
            byte[] data = IOUtils.toByteArray(request.getInputStream());
            publishRawEvent(RawConstants.KAFKA_RAW_SUB_ZIP_TOPIC, RawConstants.HTTP_STREAM_SUB_ZIP, null, data);
            return successResponse();
        } catch (Exception ex) {
            log.error("Data stream conversion failed", ex);
            return R.fail("Data stream conversion failed");
        }
    }

    /**
     * 接收传感器上传的未压缩base64数据流-整包
     *
     * @param resultMap
     * @return
     */
    @PostMapping("/unzip/upload")
    public R<Void> uploadUnZip(@RequestBody Map<String, Object> resultMap)
    {
        publishRawEvent(RawConstants.KAFKA_RAW_UNZIP_TOPIC, RawConstants.HTTP_RAW_UNZIP, resultMap, null);
        return successResponse();
    }

    /**
     * 接收传感器上传的未压缩base64数据流-分包
     *
     * @param resultMap
     * @return
     * @throws IOException
     */
    @PostMapping("/unzip/subpkg/upload")
    public R<Void> uploadSubUnZip(@RequestBody Map<String, Object> resultMap)
    {
        publishRawEvent(RawConstants.KAFKA_RAW_SUB_UNZIP_TOPIC, RawConstants.HTTP_RAW_SUB_UNZIP, resultMap, null);
        return successResponse();
    }

    /**
     * 接收上传的真实转速
     *
     * @param resultMap 原始数据
     * @return 处理结果
     */
    @PostMapping("/upload/rpm")
    public R<Void> uploadRpm(@RequestBody Map<String, Object> resultMap)
    {
        publishRawEvent(RawConstants.KAFKA_RAW_PRM_TOPIC, RawConstants.HTTP_RAW_RPM, resultMap, null);
        return successResponse();
    }

    private R<Void> successResponse()
    {
        return R.ok("Received Successful!");
    }

    private void publishRawEvent(String dataType, String source, Map<String, Object> data, byte[] byteData)
    {
        RawEvent rawEvent = new RawEvent(dataType, source, data, byteData);
        SpringUtils.context().publishEvent(rawEvent);
    }
}